package com.learn.flink.learn.time;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;

public class WatermarkDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        String hostname="192.168.1.3";
        int port = 7777;
        DataStreamSource<String> data = env.socketTextStream(hostname, port);
        SingleOutputStreamOperator<Tuple2<String, Long>> map = data.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String value) throws Exception {
                String[] split = value.split(",");
                return Tuple2.of(split[0], Long.valueOf(split[1]));
            }
        });
        //水印
        SingleOutputStreamOperator<Tuple2<String, Long>> wa = map.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Long>>() {
            Long currentMapStamp = 0L;
            Long maxOrderOut = 2000L;

            @Nullable
            @Override
            public Watermark getCurrentWatermark() {
                return new Watermark(currentMapStamp - maxOrderOut);
            }

            @Override
            public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                currentMapStamp = Math.max(currentMapStamp, element.f1);
                return element.f1;
            }
        });
        SingleOutputStreamOperator<String> result = wa.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(2))).apply(new WindowFunction<Tuple2<String, Long>, String, Tuple, TimeWindow>() {
            @Override
            public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<String> out) throws Exception {
                Iterator<Tuple2<String, Long>> iterator = input.iterator();
                List<Long> list = new ArrayList<>();
                while (iterator.hasNext()) {
                    Tuple2<String, Long> next = iterator.next();
                    list.add(next.f1);
                }
                Collections.sort(list);
                out.collect(list.get(0) + "-----" + list.get(list.size() - 1));
            }
        });
        result.print();
        env.execute();
    }
}
